package com.apache.spark.utils

import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange
import redis.clients.jedis.Jedis

import java.util
import scala.collection.mutable

/**
 * Offset管理工具类，用于往Redis中存储和读取offset
 *
 * 管理方案：
 *  1. 后置提交偏移量 -> 手动控制偏移量提交
 *     2. 手动控制偏移量提交 -> SparkStreaming提供了手动提交方案，但是我们不能用，因为我们会对DStream的结构进行转换
 *     3. 手动的提取偏移量维护到redis中
 *     -> 从kafka中消费到数据，先提取偏移量
 *     -> 等数据成功写出后，将偏移量存储到redis中
 *     -> 从kafka中消费数据之前，先到redis中读取偏移量，使用读取到的偏移量到kafka中消费数据
 *
 * 4. 手动的将偏移量存储到redis中，每次消费数据需要使用存储的offset进行消费，每次消费后，要将本次消费的offset存储到redis中
 *
 *
 */
object MyOffsetsUtils {

  /**
   * 往Redis中存储offset
   * 问题：存的offset从哪来？
   *
   * 从消费到的数据中提取出来的，传入到该方法中
   * offsetRanges: Array[OffsetRange]
   *
   * offset的结构是什么？
   *
   * Kafka中offset维护的结构
   * groupId + topic + partition => offset
   * 从传入进来的offset中提取关键信息
   *
   * 在redis怎么存？
   * 类型：string list set zset hash -> 使用hash
   * key    : groupId + topic
   * value  : partition - offset
   * 写入API : hset / hmset
   * 读取API : hgetall
   * 是否过期 : 不过期，(根据Kafka过期策略来决定)
   *
   */

  def saveOffset(topic: String, groupId: String, offsetRanges: Array[OffsetRange]): Unit = {
    //对封装的偏移量数组 offsetRanges 进行遍历
    if (offsetRanges != null && offsetRanges.length > 0) {
      val offsets: util.HashMap[String, String] = new util.HashMap[String, String]()
      for (offsetRange <- offsetRanges) {
        //获取分区
        val partition: Int = offsetRange.partition
        //获取Offset结束点
        val endOffset: Long = offsetRange.fromOffset
        offsets.put(partition.toString, endOffset.toString)
      }

      //测试提交offset
      println("提交offset : " + offsets)

      // 往redis中存
      val jedis: Jedis = MyRedisUtils.getJedisFromPool()
      val redisKey: String = s"offsets:$topic:$groupId"

      //写入
      jedis.hset(redisKey, offsets)
      //关闭
      jedis.close()
    }
  }

  /**
   * 从Redis中读取存储的offset
   * 问题:
   * 如何让SparkStreaming通过指定的offset进行消费？
   *
   * SparkStreaming要求的offset的格式是什么？
   * 格式：Map[TopicPartition,Long]
   *
   * @return 当前消费者组中，消费的主题对应的分区的偏移量信息
   */

  def readOffset(topic: String, groupId: String): Map[TopicPartition, Long] = {
    val jedis: Jedis = MyRedisUtils.getJedisFromPool()
    val redisKey: String = s"offsets:$topic:$groupId"

    //根据 key 从 Redis 中获取数据
    val offsets: util.Map[String, String] = jedis.hgetAll(redisKey)

    //测试读取offset
    println("读取到offset : " + offsets)

    val results: mutable.Map[TopicPartition, Long] = mutable.Map[TopicPartition, Long]()

    //将Java的map 转换成 Scala 的 map进行迭代
    import scala.collection.JavaConverters._
    for ((partition, offset) <- offsets.asScala) {
      val tp: TopicPartition = new TopicPartition(topic, partition.toInt)
      results.put(tp, offset.toLong)
    }
    jedis.close()

    results.toMap

  }
}
